// Copyright 2019 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: DServer support simple entity
// feature: secure discovery server and client
// feature: support auth token
// feature: provide query interfaces
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: evbs support report hostname processname and version
// feature: add statictic lookup
// feature: dds support get remote participant vbs version and rm some print
// ------------------------------------------------------------------

#include <discovery/participant/PDP.h>

#include <discovery/participant/PDPListener.h>

#include <edds/rtps/builtin/BuiltinProtocols.h>
#include <edds/rtps/builtin/liveliness/WLP.h>

#include <edds/rtps/builtin/data/ParticipantProxyData.h>
#include <history/ParticipantListener.h>
#include <deps/resources/TimedEvent.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>

#include <discovery/endpoint/EDPSimple.h>
#include <discovery/endpoint/EDPStatic.h>

#include <rtps/participant/RTPSParticipantImpl.h>

#include <edds/rtps/writer/StatelessWriter.h>
#include <edds/rtps/reader/StatelessReader.h>
#include <edds/rtps/reader/StatefulReader.h>

#include <history/WriterHistory.h>
#include <history/ReaderHistory.h>

#include <ertps/types/TypeObjectFactory.h>

#include <ertps/utils/TimeConversion.h>
#include <ertps/utils/LocatorAlias.h>

#include <ecdr/exceptions/BadParamException.h>

#if defined(_WIN32)
#include <Windows.h>
#endif

#include <mutex>
#include <chrono>

#include <edds/dds/builtin/lookup/LookupManager.hpp>
#include <rtps/builtin/data/ProxyHashTables.hpp>

#include <elog/Log.hpp>

#include <history/TopicPayloadPoolRegistry.hpp>
#include <rtps/network/ExternalLocatorsProcessor.hpp>
#include <edds/rtps/builtin/data/BuiltinEndpoints.hpp>
#include <deps/common/SystemInfo.hpp>
#include <deps/common/shared_mutex.hpp>

namespace evbs {

namespace ertps {
namespace rtps {

using namespace evbs::ertps::rtps::ddb;

// Default configuration values for PDP reliable entities.

const Duration_t pdp_heartbeat_period {0, 350U * 1000000U};         // 350 milliseconds
const Duration_t pdp_nack_response_delay {0, 100U * 1000000U};      // 100 milliseconds
const Duration_t pdp_nack_supression_duration {0, 11U * 1000000U};  // 11 milliseconds
const Duration_t pdp_heartbeat_response_delay {0, 11U * 1000000U};  // 11 milliseconds

const int32_t pdp_initial_reserved_caches = 20;

PDP::PDP(BuiltinProtocols* built, const RTPSParticipantAllocationAttributes& allocation)
    : mp_builtin(built),
      mp_RTPSParticipant(nullptr),
      mp_EDP(nullptr),
      participant_proxies_number_(allocation.participants.initial),
      participant_proxies_(allocation.participants),
      participant_deleting_proxies_(allocation.participants),
      participant_proxies_pool_(allocation.participants),
      reader_proxies_number_(allocation.total_readers().initial),
      reader_proxies_pool_(allocation.total_readers()),
      writer_proxies_number_(allocation.total_writers().initial),
      writer_proxies_pool_(allocation.total_writers()),
      m_hasChangedLocalPDP(true),
      temp_reader_proxies_({allocation.locators.max_unicast_locators, allocation.locators.max_multicast_locators,
                            allocation.data_limits}),
      temp_writer_proxies_({allocation.locators.max_unicast_locators, allocation.locators.max_multicast_locators,
                            allocation.data_limits}),
      mp_mutex(new std::recursive_mutex()),
      startTime_mutex(new std::recursive_mutex()),
      resend_participant_info_event_(nullptr) {
    size_t max_unicast_locators = allocation.locators.max_unicast_locators;
    size_t max_multicast_locators = allocation.locators.max_multicast_locators;

    for (size_t i = 0U; i < allocation.participants.initial; ++i) {
        try {
            (void)participant_proxies_pool_.push_back(new ParticipantProxyData(allocation));
        } catch (const std::exception& ex) {                                                 //LCOV_EXCL_LINE
            elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "PDP INIT ERROR: " << ex.what());  //LCOV_EXCL_LINE
        }
    }

    for (size_t i = 0U; i < allocation.total_readers().initial; ++i) {
        (void)reader_proxies_pool_.push_back(
            new ReaderProxyData(max_unicast_locators, max_multicast_locators, allocation.data_limits));
    }

    for (size_t i = 0U; i < allocation.total_writers().initial; ++i) {
        (void)writer_proxies_pool_.push_back(
            new WriterProxyData(max_unicast_locators, max_multicast_locators, allocation.data_limits));
    }
}

PDP::~PDP() {
    delete resend_participant_info_event_;
    builtin_endpoints_->disable_pdp_readers(mp_RTPSParticipant);
    delete mp_EDP;
    builtin_endpoints_->delete_pdp_endpoints(mp_RTPSParticipant);
    builtin_endpoints_.reset();

    for (ParticipantProxyData* it : participant_proxies_) {
        delete it;
    }

    for (ParticipantProxyData* it : participant_proxies_pool_) {
        delete it;
    }

    for (ReaderProxyData* it : reader_proxies_pool_) {
        delete it;
    }

    for (WriterProxyData* it : writer_proxies_pool_) {
        delete it;
    }

    delete mp_mutex;
    delete startTime_mutex;
}

ParticipantProxyData* PDP::add_participant_proxy_data(const GUID_t& participant_guid, bool with_lease_duration,
                                                      const Time_t& startTime,
                                                      const ParticipantProxyData* participant_proxy_data) {
    ParticipantProxyData* ret_val = nullptr;

    // Try to take one entry from the pool
    if (participant_proxies_pool_.empty()) {
        size_t max_proxies = participant_proxies_.max_size();
        if (participant_proxies_number_ < max_proxies) {
            // Pool is empty but limit has not been reached, so we create a new entry.
            ++participant_proxies_number_;
            try {
                ret_val = new ParticipantProxyData(mp_RTPSParticipant->getRTPSParticipantAttributes().allocation);
            } catch (std::exception& e) {                                 //LCOV_EXCL_LINE
                elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, e.what());  //LCOV_EXCL_LINE
                return nullptr;                                           //LCOV_EXCL_LINE
            }
            if (participant_guid != mp_RTPSParticipant->getGuid()) {
                ret_val->lease_duration_event = new TimedEvent(
                    mp_RTPSParticipant->getEventResource(),
                    [this, ret_val]() -> bool {
                        check_remote_participant_liveliness(ret_val);
                        return false;
                    },
                    0.0);
            }
        } else {
            logWarning(RTPS_PDP, "Maximum number of participant proxies ("
                                     << max_proxies << ") reached for participant " << mp_RTPSParticipant->getGuid()
                                     << std::endl);
            return nullptr;
        }
    } else {
        // Pool is not empty, use entry from pool
        ret_val = participant_proxies_pool_.back();
        participant_proxies_pool_.pop_back();
    }

    // Add returned entry to the collection
    ret_val->should_check_lease_duration = with_lease_duration;
    ret_val->m_guid = participant_guid;
    ret_val->m_startTime = startTime;
    ret_val->m_hostId = vbs::SystemInfo::instance().host_id();
    ret_val->m_pid = vbs::SystemInfo::instance().process_id();
    if (nullptr != participant_proxy_data) {
        ret_val->copy(*participant_proxy_data);

        ret_val->isAlive = true;
        // Notify discovery of remote participant
        getRTPSParticipant()->set_remote_evbsversion(participant_proxy_data->m_evbsRemoteVersion);
        getRTPSParticipant()->set_remote_hostname(participant_proxy_data->m_hostName);
    }
    (void)participant_proxies_.push_back(ret_val);

    std::lock_guard<std::recursive_mutex> lock(*startTime_mutex);
    (void)pdata_startTime.insert(std::make_pair(participant_guid.guidPrefix, startTime));
    return ret_val;
}

bool PDP::data_matches_with_prefix(const GuidPrefix_t& guid_prefix, const ParticipantProxyData& participant_data) {
    bool ret_val = (guid_prefix == participant_data.m_guid.guidPrefix);

#if HAVE_SECURITY
    if (!ret_val) {
        GUID_t guid = GUID_t(guid_prefix, c_EntityId_RTPSParticipant);
        return getRTPSParticipant()->security_manager().check_guid_comes_from(participant_data.m_guid, guid);
    }
#endif  // HAVE_SECURITY

    return ret_val;
}

std::string PDP::check_participant_type(const edds::dds::ParameterPropertyList_t properties) {
    auto participant_type =
        std::find_if(properties.begin(), properties.end(), [](const edds::dds::ParameterProperty_t& property) {
            return (property.first() == edds::dds::parameter_property_participant_type ||
                    property.first() == edds::dds::parameter_property_participant_type_1_0);
        });

    if (participant_type != properties.end()) {
        if (participant_type->second() == edds::rtps::ParticipantType::SERVER) {
            return edds::rtps::ParticipantType::SERVER;
        } else if (participant_type->second() == edds::rtps::ParticipantType::BACKUP) {
            return edds::rtps::ParticipantType::BACKUP;
        } else if (participant_type->second() == edds::rtps::ParticipantType::SIMPLE) {
            return edds::rtps::ParticipantType::SIMPLE;
        } else if (participant_type->second() == edds::rtps::ParticipantType::CLIENT) {
            return edds::rtps::ParticipantType::CLIENT;
        } else if (participant_type->second() == edds::rtps::ParticipantType::SUPER_CLIENT) {
            return edds::rtps::ParticipantType::SUPER_CLIENT;
        } else {
            elogError(RTPS_PDP_LISTENER, RetCode_t::RETCODE_ERROR,  //LCOV_EXCL_START
                      "Wrong " << edds::dds::parameter_property_participant_type << ": " << participant_type->second());
            return edds::rtps::ParticipantType::UNKNOWN;
        }
    } else {
        logDebug(RTPS_PDP_LISTENER, edds::dds::parameter_property_participant_type << " is not set");
        // Fallback to checking whether participant is a SERVER looking for the persistence GUID
        auto persistence_guid =
            std::find_if(properties.begin(), properties.end(), [](const edds::dds::ParameterProperty_t& property) {
                return property.first() == edds::dds::parameter_property_persistence_guid;
            });
        // The presence of persistence GUID property suggests a SERVER. This assumption is made to keep
        // backwards compatibility with Discovery Server v1.0. However, any participant that has been configured
        // as persistent will have this property.
        if (persistence_guid != properties.end()) {
            return edds::rtps::ParticipantType::SERVER;
        } else {
            return edds::rtps::ParticipantType::SIMPLE;
        }  //LCOV_EXCL_STOP
    }
}

void PDP::initializeParticipantProxyData(ParticipantProxyData* participant_data) {
    RTPSParticipantAttributes& attributes = mp_RTPSParticipant->getAttributes();

    participant_data->m_domain_id = mp_RTPSParticipant->get_domain_id();
    participant_data->m_leaseDuration = attributes.builtin.discovery_config.leaseDuration;
    // set_VendorId_vbs(participant_data->m_VendorId);
    participant_data->m_VendorId = c_VendorId_vbs;

    // TODO: participant_data->m_availableBuiltinEndpoints |= mp_builtin->available_builtin_endpoints();

    participant_data->m_availableBuiltinEndpoints |= builtin_endpoints_->builtin_endpoints();

    if (attributes.builtin.use_WriterLivelinessProtocol) {
        participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_PARTICIPANT_MESSAGE_DATA_WRITER;
        participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_PARTICIPANT_MESSAGE_DATA_READER;

#if HAVE_SECURITY
        if (mp_RTPSParticipant->is_secure()) {
            participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_PARTICIPANT_MESSAGE_SECURE_DATA_WRITER;
            participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_PARTICIPANT_MESSAGE_SECURE_DATA_READER;
        }
#endif  // if HAVE_SECURITY
    }

    if (attributes.builtin.lookup_config.use_server) {
        participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REQUEST_DATA_READER;
        participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REPLY_DATA_WRITER;
    }

    if (attributes.builtin.lookup_config.use_client) {
        participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REQUEST_DATA_WRITER;
        participant_data->m_availableBuiltinEndpoints |= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REPLY_DATA_READER;
    }
#if HAVE_SECURITY
    if (mp_RTPSParticipant->is_secure()) {
        participant_data->m_availableBuiltinEndpoints |= mp_RTPSParticipant->security_manager().builtin_endpoints();
    }
#endif  // if HAVE_SECURITY

    for (const Locator_t& loc : attributes.defaultUnicastLocatorList) {
        participant_data->default_locators.add_unicast_locator(loc);
    }
    for (const Locator_t& loc : attributes.defaultMulticastLocatorList) {
        participant_data->default_locators.add_multicast_locator(loc);
    }
    participant_data->m_guid = mp_RTPSParticipant->getGuid();
    (void)memcpy(participant_data->m_key.value, participant_data->m_guid.guidPrefix.value, 12);
    (void)memcpy(participant_data->m_key.value + 12, participant_data->m_guid.entityId.value, 4);

    // Keep persistence Guid_Prefix_t in a specific property. This info must be propagated to all
    // builtin endpoints
    {
        // Use user set persistence guid
        GuidPrefix_t persistent = mp_RTPSParticipant->get_persistence_guid_prefix();

        // If it has not been set, use guid
        if (persistent == c_GuidPrefix_Unknown) {
            persistent = attributes.prefix;
        }

        // If persistent is set, set it into the participant proxy
        if (persistent != c_GuidPrefix_Unknown) {
            participant_data->set_persistence_guid(GUID_t(persistent, c_EntityId_RTPSParticipant));
            if (participant_data->m_guid.guidPrefix.is_guid_static()) {
                participant_data->m_hostId = vbs::SystemInfo::instance().host_id();
                participant_data->m_pid = vbs::SystemInfo::instance().process_id();
            } else {
                logWarning(RTPS_PDP, "Static GUID Need set GUID Prefix to 0F.01.xx.xx.xx.xx.xx.xx.xx.xx.xx.01");
            }
        }
    }

    participant_data->metatraffic_locators.unicast.clear();
    for (const Locator_t& loc : this->mp_builtin->m_metatrafficUnicastLocatorList) {
        participant_data->metatraffic_locators.add_unicast_locator(loc);
    }

    participant_data->metatraffic_locators.multicast.clear();
    if ((!m_discovery.avoid_builtin_multicast) || participant_data->metatraffic_locators.unicast.empty()) {
        for (const Locator_t& loc : this->mp_builtin->m_metatrafficMulticastLocatorList) {
            participant_data->metatraffic_locators.add_multicast_locator(loc);
        }
    }

    edds::rtps::ExternalLocatorsProcessor::add_external_locators(
        *participant_data, attributes.builtin.metatraffic_external_unicast_locators,
        attributes.default_external_unicast_locators);

    participant_data->m_participantName = std::string(attributes.getName());
    participant_data->m_hostName = vbs::SystemInfo::get_hostname();

#if HAVE_SECURITY
    if (mp_RTPSParticipant->is_secure()) {
        IdentityToken* identity_token = nullptr;
        if (mp_RTPSParticipant->security_manager().get_identity_token(&identity_token) && (identity_token != nullptr)) {
            participant_data->identity_token_ = std::move(*identity_token);
            (void)mp_RTPSParticipant->security_manager().return_identity_token(identity_token);
        }

        PermissionsToken* permissions_token = nullptr;
        if (mp_RTPSParticipant->security_manager().get_permissions_token(&permissions_token) &&
            (permissions_token != nullptr)) {
            participant_data->permissions_token_ = std::move(*permissions_token);
            (void)mp_RTPSParticipant->security_manager().return_permissions_token(permissions_token);
        }
        const security::ParticipantSecurityAttributes& sec_attrs = mp_RTPSParticipant->security_attributes();
        participant_data->security_attributes_ = sec_attrs.mask();
        participant_data->plugin_security_attributes_ = sec_attrs.plugin_participant_attributes;
    } else {
        participant_data->security_attributes_ = 0UL;
        participant_data->plugin_security_attributes_ = 0UL;
    }
    participant_data->m_auth_token = mp_RTPSParticipant->auth_token();
    logDebug(AUTHTOKEN,
             "localpartproxytoken:" << participant_data->m_guid << ":" << mp_RTPSParticipant->auth_token().authToken());
#endif  // if HAVE_SECURITY

    // get particpant init time
    participant_data->m_startTime.now(participant_data->m_startTime);

    // Set properties that will be sent to Proxy Data
    set_external_participant_properties_(participant_data);

    participant_data->transport_type = attributes.transport_type;
    if (attributes.builtin.discovery_config.is_compatible) {
        participant_data->isCompatible = true;
    }
}

bool PDP::initPDP(RTPSParticipantImpl* part) {
    mp_RTPSParticipant = part;
    m_discovery = mp_RTPSParticipant->getAttributes().builtin;
    initial_announcements_ = m_discovery.discovery_config.initial_announcements;
    // CREATE ENDPOINTS
    if (!createPDPEndpoints()) {
        return false;
    }

    // UPDATE METATRAFFIC.
    update_builtin_locators();

    ParticipantProxyData* pdata = nullptr;
    {
        std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
        Time_t startTime(0, 0U);
        pdata = add_participant_proxy_data(mp_RTPSParticipant->getGuid(), false, startTime, nullptr);
    }

    if (pdata == nullptr) {
        return false;
    }
    initializeParticipantProxyData(pdata);

    std::string part_type = check_participant_type(pdata->m_properties);
    if (part_type == edds::rtps::ParticipantType::SERVER || part_type == edds::rtps::ParticipantType::BACKUP) {
        pdata->m_domain_id = DOMAIN_ID_UNKNOWN;
    }
    return true;
}

bool PDP::enable() {
    // It is safe to call enable() on already enable PDPs
    if (enabled_) {
        return true;
    }

    pre_enable_actions();

    // Create lease events on already created proxy data objects
    for (ParticipantProxyData* pool_item : participant_proxies_pool_) {
        pool_item->lease_duration_event = new TimedEvent(
            mp_RTPSParticipant->getEventResource(),
            [this, pool_item]() -> bool {
                check_remote_participant_liveliness(pool_item);
                return false;
            },
            0.0);
    }

    resend_participant_info_event_ = new TimedEvent(
        mp_RTPSParticipant->getEventResource(),
        [&]() -> bool {
            announceParticipantState(false);
            set_next_announcement_interval();
            return true;
        },
        0);

    set_initial_announcement_interval();

    enabled_.store(true);

    return builtin_endpoints_->enable_pdp_readers(mp_RTPSParticipant);
}

void PDP::pre_enable_actions() {
    // Empty implementation
}

void PDP::disable() {
    // Extract all the participant proxies excluding first one (ourselves)
    std::vector<ParticipantProxyData*> participants;
    {
        std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
        participants.insert(participants.end(), participant_proxies_.begin() + 1, participant_proxies_.end());
        participant_proxies_.erase(participant_proxies_.begin() + 1, participant_proxies_.end());
    }

    // Unmatch all remote participants
    for (ParticipantProxyData* pdata : participants) {
        actions_on_remote_participant_removed(pdata, pdata->m_guid, ParticipantDiscoveryInfo::REMOVED_PARTICIPANT,
                                              nullptr);
    }
}

void PDP::announceParticipantState(bool new_change, bool dispose /* = false */) {
    WriteParams __wp = WriteParams::write_params_default();
    announceParticipantState(new_change, dispose, __wp);
}

void PDP::announceParticipantState(WriterHistory& history, bool new_change, bool dispose, WriteParams& wparams) {
    if (enabled_) {
        CacheChange_t* change = nullptr;
        if (!dispose) {
            logDebug(RTPS_PDP, "announceParticipantState alive");
            if (m_hasChangedLocalPDP.exchange(false) || new_change) {
                mp_mutex->lock();
                ParticipantProxyData* local_participant_data = getLocalParticipantProxyData();
                InstanceHandle_t key = local_participant_data->m_key;
                ParticipantProxyData proxy_data_copy(*local_participant_data);
                mp_mutex->unlock();

                if (history.getHistorySize() > 0U) {
                    (void)history.remove_min_change();
                }

                uint32_t cdr_size = proxy_data_copy.get_serialized_size(true);
                change = history.new_change([cdr_size]() -> uint32_t { return cdr_size; }, ALIVE, key);

                if (change != nullptr) {
                    CDRMessage_t aux_msg(change->serializedPayload);

#if __BIG_ENDIAN__
                    change->serializedPayload.encapsulation = (uint16_t)PL_CDR_BE;
                    aux_msg.msg_endian = BIGEND;
#else
                    change->serializedPayload.encapsulation = (uint16_t)PL_CDR_LE;
                    aux_msg.msg_endian = LITTLEEND;
#endif  // if __BIG_ENDIAN__
                    if (proxy_data_copy.writeToCDRMessage(&aux_msg, true)) {
                        change->serializedPayload.length = (uint16_t)aux_msg.length;
                        (void)history.add_change(change, wparams);
                    } else {
                        elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "Cannot serialize ParticipantProxyData.");
                    }
                }
            }

        } else {
            logDebug(RTPS_PDP, "announceParticipantState dispose");

            mp_mutex->lock();
            ParticipantProxyData proxy_data_copy(*getLocalParticipantProxyData());
            mp_mutex->unlock();

            if (history.getHistorySize() > 0U) {
                (void)history.remove_min_change();
            }

            uint32_t cdr_size = proxy_data_copy.get_serialized_size(true);
            change = history.new_change([cdr_size]() -> uint32_t { return cdr_size; }, NOT_ALIVE_DISPOSED_UNREGISTERED,
                                        getLocalParticipantProxyData()->m_key);

            if (change != nullptr) {
                CDRMessage_t aux_msg(change->serializedPayload);

#if __BIG_ENDIAN__
                change->serializedPayload.encapsulation = (uint16_t)PL_CDR_BE;
                aux_msg.msg_endian = BIGEND;
#else
                change->serializedPayload.encapsulation = (uint16_t)PL_CDR_LE;
                aux_msg.msg_endian = LITTLEEND;
#endif  // if __BIG_ENDIAN__

                if (proxy_data_copy.writeToCDRMessage(&aux_msg, true)) {
                    change->serializedPayload.length = (uint16_t)aux_msg.length;
                    (void)history.add_change(change, wparams);
                } else {
                    elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "Cannot serialize ParticipantProxyData.");
                }
            }
        }
    }
}

void PDP::stopParticipantAnnouncement() {
    if (resend_participant_info_event_) {
        resend_participant_info_event_->cancel_timer();
    }
}

void PDP::resetParticipantAnnouncement() {
    if (resend_participant_info_event_) {
        resend_participant_info_event_->restart_timer();
    }
}

void PDP::notify_and_maybe_ignore_new_participant(ParticipantProxyData* pdata, bool& should_be_ignored) {
    should_be_ignored = false;
    logDebug(RTPS_PDP_DISCOVERY, "New participant " << pdata->m_guid << " at "
                                                    << "MTTLoc: " << pdata->metatraffic_locators
                                                    << " DefLoc:" << pdata->default_locators);

    vbs::common::ParticipantListener* listener = getRTPSParticipant()->getListener();
    if (listener) {
        {
            std::lock_guard<std::mutex> cb_lock(callback_mtx_);
            ParticipantDiscoveryInfo info(*pdata);
            info.status = ParticipantDiscoveryInfo::DISCOVERED_PARTICIPANT;
            listener->onParticipantDiscovery(getRTPSParticipant()->getUserRTPSParticipant(), std::move(info));
        }

        if (should_be_ignored) {
            getRTPSParticipant()->ignore_participant(pdata->m_guid.guidPrefix);
        }
    }
}

void PDP::start_initial_announcement(uint32_t count, Duration_t period) {
    initial_announcements_.count = count;
    initial_announcements_.period = period;
    set_initial_announcement_interval();
}

//! get is_send_multi
bool PDP::getSendMulti() const {
    return builtin_endpoints_->getSendMulti();
}

//! set is_send_multi
void PDP::setSendMulti(bool send_multi) {
    return builtin_endpoints_->setSendMulti(send_multi);
}

bool PDP::has_reader_proxy_data(const GUID_t& reader) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == reader.guidPrefix) {
            ProxyHashTable<ReaderProxyData>& readers = *pit->m_readers;
            return readers.find(reader.entityId) != readers.end();
        }
    }
    return false;
}

bool PDP::lookupReaderProxyData(const GUID_t& reader, ReaderProxyData& rdata) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == reader.guidPrefix) {
            auto rit = pit->m_readers->find(reader.entityId);
            if (rit != pit->m_readers->end()) {
                rdata.copy(rit->second);
                return true;
            }
        }
    }
    return false;
}

bool PDP::has_writer_proxy_data(const GUID_t& writer) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == writer.guidPrefix) {
            ProxyHashTable<WriterProxyData>& writers = *pit->m_writers;
            return writers.find(writer.entityId) != writers.end();
        }
    }
    return false;
}

bool PDP::lookupWriterProxyData(const GUID_t& writer, WriterProxyData& wdata) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == writer.guidPrefix) {
            auto wit = pit->m_writers->find(writer.entityId);
            if (wit != pit->m_writers->end()) {
                wdata.copy(wit->second);
                return true;
            }
        }
    }
    return false;
}

bool PDP::removeReaderProxyData(const GUID_t& reader_guid) {
    logDebug(RTPS_PDP, "Removing reader proxy data " << reader_guid);
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);

    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == reader_guid.guidPrefix) {
            auto rit = pit->m_readers->find(reader_guid.entityId);

            if (rit != pit->m_readers->end()) {
                ReaderProxyData* pR = rit->second;
                (void)mp_EDP->unpairReaderProxy(pit->m_guid, reader_guid);

                vbs::common::ParticipantListener* listener = mp_RTPSParticipant->getListener();
                if (listener) {
                    ReaderDiscoveryInfo info(std::move(*pR));
                    info.status = READER_DISCOVERY_STATUS::REMOVED_READER;
                    listener->onReaderDiscovery(mp_RTPSParticipant->getUserRTPSParticipant(), std::move(info));
                }

                // Clear reader proxy data and move to pool in order to allow reuse
                pR->clear();
                (void)pit->m_readers->erase(rit);
                (void)reader_proxies_pool_.push_back(pR);
                return true;
            }
        }
    }

    return false;
}

bool PDP::removeWriterProxyData(const GUID_t& writer_guid) {
    try {
        logDebug(RTPS_PDP, "Removing writer proxy data " << writer_guid);
    } catch (...) {
        return false;
    }
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);

    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == writer_guid.guidPrefix) {
            auto wit = pit->m_writers->find(writer_guid.entityId);

            if (wit != pit->m_writers->end()) {
                WriterProxyData* pW = wit->second;
                (void)mp_EDP->unpairWriterProxy(pit->m_guid, writer_guid, false);

                vbs::common::ParticipantListener* listener = mp_RTPSParticipant->getListener();
                if (listener) {
                    WriterDiscoveryInfo info(std::move(*pW));
                    info.status = WRITER_DISCOVERY_STATUS::REMOVED_WRITER;
                    listener->onWriterDiscovery(mp_RTPSParticipant->getUserRTPSParticipant(), std::move(info));
                }

                // Clear writer proxy data and move to pool in order to allow reuse
                pW->clear();
                (void)pit->m_writers->erase(wit);
                (void)writer_proxies_pool_.push_back(pW);

                return true;
            }
        }
    }

    return false;
}

bool PDP::lookup_participant_name(const GUID_t& guid, string_255& name) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid == guid) {
            name = pit->m_participantName;
            return true;
        }
    }
    return false;
}

bool PDP::lookup_participant_key(const GUID_t& participant_guid, InstanceHandle_t& key) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid == participant_guid) {
            key = pit->m_key;
            return true;
        }
    }
    return false;
}

bool PDP::lookup_participant_locators(const GuidPrefix_t& guid_prefix, LocatorList_t& locators) {
    std::lock_guard<std::recursive_mutex> guardPDP(*this->mp_mutex);
    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == guid_prefix) {
            for (const Locator_t& locator : pit->metatraffic_locators.unicast) {
                locators.push_back(locator);
            }
            return true;
        }
    }
    return false;
}

ReaderProxyData* PDP::addReaderProxyData(
    const GUID_t& reader_guid, GUID_t& participant_guid,
    std::function<bool(ReaderProxyData*, bool, const ParticipantProxyData&, RTPSParticipantImpl*)> initializer_func) {
    logDebug(RTPS_PDP, "Adding reader proxy data " << reader_guid);
    ReaderProxyData* ret_val = nullptr;

    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);

    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == reader_guid.guidPrefix) {
            // Copy participant data to be used outside.
            participant_guid = pit->m_guid;

            // Check that it is not already there:
            auto rpi = pit->m_readers->find(reader_guid.entityId);

            if (rpi != pit->m_readers->end()) {
                ret_val = rpi->second;

                if (!initializer_func(ret_val, true, *pit, mp_RTPSParticipant)) {
                    return nullptr;
                }

                vbs::common::ParticipantListener* listener = mp_RTPSParticipant->getListener();
                if (listener) {
                    ReaderDiscoveryInfo info(*ret_val);
                    info.status = READER_DISCOVERY_STATUS::CHANGED_QOS_READER;
                    listener->onReaderDiscovery(mp_RTPSParticipant->getUserRTPSParticipant(), std::move(info));
                    check_and_notify_type_discovery(listener, *ret_val);
                }

                return ret_val;
            }

            // Try to take one entry from the pool
            if (reader_proxies_pool_.empty()) {
                size_t max_proxies = reader_proxies_pool_.max_size();
                if (reader_proxies_number_ < max_proxies) {
                    // Pool is empty but limit has not been reached, so we create a new entry.
                    ++reader_proxies_number_;
                    ret_val = new ReaderProxyData(
                        mp_RTPSParticipant->getAttributes().allocation.locators.max_unicast_locators,
                        mp_RTPSParticipant->getAttributes().allocation.locators.max_multicast_locators,
                        mp_RTPSParticipant->getAttributes().allocation.data_limits);
                } else {
                    logWarning(RTPS_PDP, "Maximum number of reader proxies ("
                                             << max_proxies << ") reached for participant "
                                             << mp_RTPSParticipant->getGuid() << std::endl);
                    return nullptr;
                }
            } else {
                // Pool is not empty, use entry from pool
                ret_val = reader_proxies_pool_.back();
                reader_proxies_pool_.pop_back();
            }
            try {
                // Copy network configuration from participant to reader proxy
                // ret_val->networkConfiguration(pit->m_networkConfiguration);
                // Add to ParticipantProxyData
                ret_val->set_host_id(pit->m_hostId);
                ret_val->set_pid(pit->m_pid);
                (*pit->m_readers)[reader_guid.entityId] = ret_val;
            } catch (const foonathan::memory::bad_node_size& e) {  //LCOV_EXCL_START
                logWarning(RTPS_PDP, std::string("catch bad_node_size exception:") << e.what());
                return nullptr;
            } catch (const foonathan::memory::bad_alignment& e) {
                logWarning(RTPS_PDP, std::string("catch bad_alignment exception:") << e.what());
                return nullptr;
            } catch (const std::exception& e) {
                logWarning(RTPS_PDP, std::string("catch other exception:") << e.what());
                return nullptr;  //LCOV_EXCL_STOP
            }

            if (!initializer_func(ret_val, false, *pit, mp_RTPSParticipant)) {
                return nullptr;
            }

            vbs::common::ParticipantListener* listener = mp_RTPSParticipant->getListener();
            if (listener) {
                ReaderDiscoveryInfo info(*ret_val);
                info.status = READER_DISCOVERY_STATUS::DISCOVERED_READER;
                listener->onReaderDiscovery(mp_RTPSParticipant->getUserRTPSParticipant(), std::move(info));
                try {
                    check_and_notify_type_discovery(listener, *ret_val);
                } catch (const std::range_error& e) {  //LCOV_EXCL_START
                    logWarning(RTPS_PDP, std::string("catch range_error exception:") << e.what());
                    return nullptr;
                } catch (const vbsutil::ecdr::exception::BadParamException& e) {
                    logWarning(RTPS_PDP, std::string("catch BadParamException exception:") << e.what());
                    return nullptr;
                } catch (const std::exception& e) {
                    logWarning(RTPS_PDP, std::string("catch other exception:") << e.what());
                    return nullptr;  //LCOV_EXCL_STOP
                }
            }

            return ret_val;
        }
    }

    return nullptr;
}

WriterProxyData* PDP::addWriterProxyData(
    const GUID_t& writer_guid, GUID_t& participant_guid,
    std::function<bool(WriterProxyData*, bool, const ParticipantProxyData&, RTPSParticipantImpl*)> initializer_func) {
    logDebug(RTPS_PDP, "Adding writer proxy data " << writer_guid);
    WriterProxyData* ret_val = nullptr;

    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);

    for (ParticipantProxyData* pit : participant_proxies_) {
        if (pit->m_guid.guidPrefix == writer_guid.guidPrefix) {
            // Copy participant data to be used outside.
            participant_guid = pit->m_guid;

            // Check that it is not already there:
            auto wpi = pit->m_writers->find(writer_guid.entityId);

            if (wpi != pit->m_writers->end()) {
                ret_val = wpi->second;

                if (!initializer_func(ret_val, true, *pit, mp_RTPSParticipant)) {
                    return nullptr;
                }

                vbs::common::ParticipantListener* listener = mp_RTPSParticipant->getListener();
                if (listener) {
                    WriterDiscoveryInfo info(*ret_val);
                    info.status = WRITER_DISCOVERY_STATUS::CHANGED_QOS_WRITER;
                    listener->onWriterDiscovery(mp_RTPSParticipant->getUserRTPSParticipant(), std::move(info));
                    check_and_notify_type_discovery(listener, *ret_val);
                }

                return ret_val;
            }

            // Try to take one entry from the pool
            if (writer_proxies_pool_.empty()) {
                size_t max_proxies = writer_proxies_pool_.max_size();
                if (writer_proxies_number_ < max_proxies) {
                    // Pool is empty but limit has not been reached, so we create a new entry.
                    ++writer_proxies_number_;
                    auto allocations = mp_RTPSParticipant->getAttributes().allocation;
                    ret_val = new WriterProxyData(allocations.locators.max_unicast_locators,
                                                  allocations.locators.max_multicast_locators, allocations.data_limits);
                } else {
                    logWarning(RTPS_PDP, "Maximum number of writer proxies ("
                                             << max_proxies << ") reached for participant "
                                             << mp_RTPSParticipant->getGuid() << std::endl);
                    return nullptr;
                }
            } else {
                // Pool is not empty, use entry from pool
                ret_val = writer_proxies_pool_.back();
                writer_proxies_pool_.pop_back();
            }

            try {
                // Copy network configuration from participant to writer proxy
                // ret_val->networkConfiguration(pit->m_networkConfiguration);
                // Add to ParticipantProxyData
                ret_val->set_host_id(pit->m_hostId);
                ret_val->set_pid(pit->m_pid);
                (*pit->m_writers)[writer_guid.entityId] = ret_val;
            } catch (const foonathan::memory::bad_node_size& e) {  //LCOV_EXCL_START
                logWarning(RTPS_PDP, std::string("catch bad_node_size exception:") << e.what());
                return nullptr;
            } catch (const foonathan::memory::bad_alignment& e) {
                logWarning(RTPS_PDP, std::string("catch bad_alignment exception:") << e.what());
                return nullptr;
            } catch (const foonathan::memory::out_of_memory& e) {
                logWarning(RTPS_PDP, std::string("catch out_of_memory exception:") << e.what());
                return nullptr;
            } catch (const std::exception& e) {
                logWarning(RTPS_PDP, "addWriterProxyData() catch other error: " << e.what());
                return nullptr;
            }  //LCOV_EXCL_STOP

            if (!initializer_func(ret_val, false, *pit, mp_RTPSParticipant)) {
                return nullptr;
            }

            vbs::common::ParticipantListener* listener = mp_RTPSParticipant->getListener();
            if (listener) {
                WriterDiscoveryInfo info(*ret_val);
                info.status = WRITER_DISCOVERY_STATUS::DISCOVERED_WRITER;
                listener->onWriterDiscovery(mp_RTPSParticipant->getUserRTPSParticipant(), std::move(info));
                try {
                    check_and_notify_type_discovery(listener, *ret_val);
                } catch (const std::range_error& e) {  //LCOV_EXCL_START
                    logWarning(RTPS_PDP, "catch range_error error: " << e.what());
                }  //LCOV_EXCL_STOP
            }

            return ret_val;
        }
    }

    return nullptr;
}

#if HAVE_SECURITY
bool PDP::pairing_remote_writer_with_local_reader_after_security(const GUID_t& local_reader,
                                                                 const WriterProxyData& remote_writer_data) {
    return mp_EDP->pairing_remote_writer_with_local_reader_after_security(local_reader, remote_writer_data);
}

bool PDP::pairing_remote_reader_with_local_writer_after_security(const GUID_t& local_writer,
                                                                 const ReaderProxyData& remote_reader_data) {
    return mp_EDP->pairing_remote_reader_with_local_writer_after_security(local_writer, remote_reader_data);
}

#endif  // HAVE_SECURITY

bool PDP::remove_remote_participant(const GUID_t& partGUID, ParticipantDiscoveryInfo::DISCOVERY_STATUS reason) {
    if (partGUID == getLocalParticipantProxyData()->m_guid) {
        // avoid removing our own data
        return false;
    }

    logDebug(RTPS_PDP, partGUID);
    ParticipantProxyData* pdata = nullptr;

    // Remove it from our vector or RTPSParticipantProxies:
    {
        std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
        for (ResourceLimitedVector<ParticipantProxyData*>::iterator pit = participant_proxies_.begin();
             pit != participant_proxies_.end(); ++pit) {
            if ((*pit)->m_guid == partGUID) {
                pdata = *pit;
                (void)participant_deleting_proxies_.push_back(pdata);
                (void)participant_proxies_.erase(pit);
                break;
            }
        }
        std::lock_guard<std::recursive_mutex> lock(*startTime_mutex);
        pdata_startTime.erase(partGUID.guidPrefix);
    }

    if (pdata != nullptr) {
        vbs::common::ParticipantListener* listener = mp_RTPSParticipant->getListener();
        actions_on_remote_participant_removed(pdata, partGUID, reason, listener);
        return true;
    }

    return false;
}

void PDP::remove_remote_all_entities(ParticipantProxyData* pdata, const GUID_t& partGUID,
                                     ParticipantDiscoveryInfo::DISCOVERY_STATUS reason, ParticipantListener* listener) {
    assert(nullptr != pdata);
    if (mp_EDP != nullptr) {
        for (auto pit : *pdata->m_readers) {
            ReaderProxyData* rit = pit.second;
            GUID_t reader_guid(rit->guid());
            if (reader_guid != c_Guid_Unknown) {
                mp_EDP->unpairReaderProxy(partGUID, reader_guid);

                if (listener) {
                    RTPSParticipant* participant = mp_RTPSParticipant->getUserRTPSParticipant();
                    ReaderDiscoveryInfo info(std::move(*rit));
                    info.status = READER_DISCOVERY_STATUS::REMOVED_READER;
                    listener->onReaderDiscovery(participant, std::move(info));
                }
            }
        }
        for (auto pit : *pdata->m_writers) {
            WriterProxyData* wit = pit.second;
            GUID_t writer_guid(wit->guid());
            if (writer_guid != c_Guid_Unknown) {
                mp_EDP->unpairWriterProxy(partGUID, writer_guid,
                                          reason == ParticipantDiscoveryInfo::DROPPED_PARTICIPANT);

                if (listener) {
                    RTPSParticipant* participant = mp_RTPSParticipant->getUserRTPSParticipant();
                    WriterDiscoveryInfo info(std::move(*wit));
                    info.status = WRITER_DISCOVERY_STATUS::REMOVED_WRITER;
                    listener->onWriterDiscovery(participant, std::move(info));
                }
            }
        }
    }

    if (mp_builtin->mp_WLP != nullptr) {
        mp_builtin->mp_WLP->removeRemoteEndpoints(pdata);
    }

    if (mp_builtin->tlm_ != nullptr) {
        mp_builtin->tlm_->remove_remote_endpoints(pdata);
    }

    if (mp_EDP != nullptr) {
        mp_EDP->removeRemoteEndpoints(pdata);
    }

    removeRemoteEndpoints(pdata);

#if HAVE_SECURITY
    mp_builtin->mp_participantImpl->security_manager().remove_participant(*pdata);
#endif  // if HAVE_SECURITY
}

void PDP::actions_on_remote_participant_removed(ParticipantProxyData* pdata, const GUID_t& partGUID,
                                                ParticipantDiscoveryInfo::DISCOVERY_STATUS reason,
                                                vbs::common::ParticipantListener* listener) {
    assert(nullptr != pdata);
    if (mp_EDP != nullptr) {
        for (auto pit : *pdata->m_readers) {
            ReaderProxyData* rit = pit.second;
            GUID_t reader_guid(rit->guid());
            if (reader_guid != c_Guid_Unknown) {
                mp_EDP->unpairReaderProxy(partGUID, reader_guid);

                if (listener) {
                    RTPSParticipant* participant = mp_RTPSParticipant->getUserRTPSParticipant();
                    ReaderDiscoveryInfo info(std::move(*rit));
                    info.status = READER_DISCOVERY_STATUS::REMOVED_READER;
                    listener->onReaderDiscovery(participant, std::move(info));
                }
            }
        }
        for (auto pit : *pdata->m_writers) {
            WriterProxyData* wit = pit.second;
            GUID_t writer_guid(wit->guid());
            if (writer_guid != c_Guid_Unknown) {
                mp_EDP->unpairWriterProxy(partGUID, writer_guid,
                                          reason == ParticipantDiscoveryInfo::DROPPED_PARTICIPANT);

                if (listener) {
                    RTPSParticipant* participant = mp_RTPSParticipant->getUserRTPSParticipant();
                    WriterDiscoveryInfo info(std::move(*wit));
                    info.status = WRITER_DISCOVERY_STATUS::REMOVED_WRITER;
                    listener->onWriterDiscovery(participant, std::move(info));
                }
            }
        }
    }

    if (mp_builtin->mp_WLP != nullptr) {
        mp_builtin->mp_WLP->removeRemoteEndpoints(pdata);
    }

    if (mp_builtin->tlm_ != nullptr) {
        mp_builtin->tlm_->remove_remote_endpoints(pdata);
    }

    if (mp_EDP != nullptr) {
        mp_EDP->removeRemoteEndpoints(pdata);
    }

    removeRemoteEndpoints(pdata);

#if HAVE_SECURITY
    mp_builtin->mp_participantImpl->security_manager().remove_participant(*pdata);
#endif  // if HAVE_SECURITY

    builtin_endpoints_->remove_from_pdp_reader_history(pdata->m_key);

    // auto listener = mp_RTPSParticipant->getListener();
    if (listener != nullptr) {
        std::lock_guard<std::mutex> lock(callback_mtx_);
        ParticipantDiscoveryInfo info(*pdata);
        info.status = reason;
        listener->onParticipantDiscovery(mp_RTPSParticipant->getUserRTPSParticipant(), std::move(info));
    }

    {
        std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);

        // Return reader proxy objects to pool
        for (auto pit : *pdata->m_readers) {
            pit.second->clear();
            reader_proxies_pool_.push_back(pit.second);
        }
        pdata->m_readers->clear();

        // Return writer proxy objects to pool
        for (auto pit : *pdata->m_writers) {
            pit.second->clear();
            writer_proxies_pool_.push_back(pit.second);
        }
        pdata->m_writers->clear();

        // Cancel lease event
        if (nullptr != pdata->lease_duration_event) {
            pdata->lease_duration_event->cancel_timer();
        }

        for (ResourceLimitedVector<ParticipantProxyData*>::iterator pit = participant_deleting_proxies_.begin();
             pit != participant_deleting_proxies_.end(); ++pit) {
            if ((*pit)->m_guid == partGUID) {
                pdata = *pit;
                (void)participant_deleting_proxies_.erase(pit);
                break;
            }
        }
        // Return proxy object to pool
        pdata->clear();
        participant_proxies_pool_.push_back(pdata);
    }
}

const BuiltinAttributes& PDP::builtin_attributes() const {
    return mp_builtin->m_att;
}

void PDP::assert_remote_participant_liveliness(const GuidPrefix_t& remote_guid) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);

    for (ParticipantProxyData* it : this->participant_proxies_) {
        if (it->m_guid.guidPrefix == remote_guid) {
            // TODO Ricardo: Study if isAlive attribute is necessary.
            it->isAlive = true;
            it->assert_liveliness();
            break;
        }
    }
}

CDRMessage_t PDP::get_participant_proxy_data_serialized(Endianness_t endian) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
    CDRMessage_t cdr_msg(RTPSMESSAGE_DEFAULT_SIZE);
    cdr_msg.msg_endian = endian;

    if (!getLocalParticipantProxyData()->writeToCDRMessage(&cdr_msg, false)) {
        cdr_msg.pos = 0U;
        cdr_msg.length = 0U;
    }

    return cdr_msg;
}

ParticipantProxyData* PDP::get_participant_proxy_data(const GuidPrefix_t& guid_prefix) {
    for (auto pit = ParticipantProxiesBegin(); pit != ParticipantProxiesEnd(); ++pit) {
        if (data_matches_with_prefix(guid_prefix, **pit)) {
            return *(pit);
        }
    }
    return nullptr;
}

EvbsVersion_t PDP::get_version_by_guid(const GuidPrefix_t& guid_prefix) {
    std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);

    for (auto pit = ParticipantProxiesBegin(); pit != ParticipantProxiesEnd(); ++pit) {
        if (guid_prefix == (*pit)->m_guid.guidPrefix) {
            return (*pit)->m_evbsRemoteVersion;
        }
    }
    return evbs::edds::rtps::c_Version_vbs_unknown;
}

evbs::edds::rtps::LocatorList& PDP::remote_server_locators() {
    return mp_builtin->m_DiscoveryServers;
}

void PDP::check_remote_participant_liveliness(ParticipantProxyData* remote_participant) {
    std::unique_lock<std::recursive_mutex> guard(*mp_mutex);

    if (remote_participant->should_check_lease_duration) {
        assert(GUID_t::unknown() != remote_participant->m_guid);
        // Check last received message's time_point plus lease duration time doesn't overcome now().
        // If overcame, remove participant.
        auto now = std::chrono::steady_clock::now();
        auto real_lease_tm =
            remote_participant->last_received_message_tm() +
            std::chrono::microseconds(TimeConv::Duration_t2MicroSecondsInt64(remote_participant->m_leaseDuration));
        if (now > real_lease_tm) {
            std::vector<GUID_t> tmp_client_by_server;
            GUID_t Server_guid = GUID_t::unknown();
            if (getLocalParticipantProxyData()->get_participant_type() == edds::rtps::ParticipantType::SERVER &&
                remote_participant->get_participant_type() == edds::rtps::ParticipantType::SERVER) {
                Server_guid = remote_participant->m_guid;
                for (auto it = client_by_server.begin(); it != client_by_server.end();) {
                    if (it->second == remote_participant->m_guid.guidPrefix) {
                        tmp_client_by_server.push_back(it->first);
                        it = client_by_server.erase(it);
                    } else {
                        ++it;
                    }
                }
            }
            guard.unlock();
            logWarning(RTPS_PDP, "Participant " << remote_participant->m_guid << " timeout and removed.");
            remove_monitored_guid(remote_participant->m_guid);
            (void)remove_remote_participant(remote_participant->m_guid, ParticipantDiscoveryInfo::DROPPED_PARTICIPANT);
            for (auto it : tmp_client_by_server) {
                // The remote server goes offline, and the client it synchronizes with goes offline at the same time.
                logInfo(RTPS_PDP,
                        "Client " << it << " is removed due to Server " << Server_guid << " timeout and offline");
                (void)remove_remote_participant(it, ParticipantDiscoveryInfo::DROPPED_PARTICIPANT);
            }
            return;
        }

        // Calculate next trigger.
        auto next_trigger = real_lease_tm - now;
        (void)remote_participant->lease_duration_event->update_interval_millisec(
            (double_t)std::chrono::duration_cast<std::chrono::milliseconds>(next_trigger).count());
        remote_participant->lease_duration_event->restart_timer();
    }
}

void PDP::check_and_notify_type_discovery(vbs::common::ParticipantListener*, const WriterProxyData& wdata) const {
    edds::dds::builtin::LookupManager* tlm = mp_RTPSParticipant->getUserRTPSParticipant()->lookup_manager();
    auto type_name = wdata.typeName().to_string();
    if (tlm && tlm->type_has_registered(type_name)) {
        tlm->get_type_info(type_name);
        tlm->restart_timer();
    }
}

void PDP::check_and_notify_type_discovery(vbs::common::ParticipantListener*, const ReaderProxyData& rdata) const {
    edds::dds::builtin::LookupManager* tlm = mp_RTPSParticipant->getUserRTPSParticipant()->lookup_manager();
    auto type_name = rdata.typeName().to_string();
    if (tlm && tlm->type_has_registered(type_name)) {
        tlm->get_type_info(type_name);
        tlm->restart_timer();
    }
}

void PDP::set_next_announcement_interval() {
    if (initial_announcements_.count > 0U) {
        --initial_announcements_.count;
        (void)resend_participant_info_event_->update_interval(initial_announcements_.period);
    } else {
        (void)resend_participant_info_event_->update_interval(
            m_discovery.discovery_config.leaseDuration_announcementperiod);
    }
}

void PDP::set_initial_announcement_interval() {
    if ((initial_announcements_.count > 0U) && (initial_announcements_.period <= c_TimeZero)) {
        // Force a small interval (1ms) between initial announcements
        logWarning(RTPS_PDP,
                   "Initial announcement period is not strictly positive. Changing to 1ms.");  //LCOV_EXCL_START
        initial_announcements_.period = {0, 1000000U};
    }  //LCOV_EXCL_STOP
    set_next_announcement_interval();
}

void PDP::resend_ininitial_announcements() {
    if (enabled_) {
        {
            std::lock_guard<std::recursive_mutex> guardPDP(*mp_mutex);
            initial_announcements_ = m_discovery.discovery_config.initial_announcements;
        }
        set_next_announcement_interval();
        resetParticipantAnnouncement();
    }
}

void PDP::set_external_participant_properties_(ParticipantProxyData* participant_data) {
    // For each property add it if it should be sent (it is propagated)
    for (auto const& property : mp_RTPSParticipant->getAttributes().properties.properties()) {
        if (property.propagate()) {
            (void)participant_data->m_properties.push_back(property.name(), property.value());
        }
    }

    // Set participant type property
    // TODO: This could be done somewhere else that makes more sense.
    std::stringstream participant_type;
    participant_type << mp_RTPSParticipant->getAttributes().builtin.discovery_config.discoveryProtocol;
    auto ptype = participant_type.str();
    (void)participant_data->m_properties.push_back(edds::dds::parameter_property_participant_type, ptype);

    // Add physical properties if present
    // TODO: This should be done using propagate value, however this cannot be done without breaking
    // compatibility
    std::vector<std::string> physical_property_names = {edds::dds::parameter_policy_physical_data_host,
                                                        edds::dds::parameter_policy_physical_data_user,
                                                        edds::dds::parameter_policy_physical_data_process};
    for (auto& physical_property_name : physical_property_names) {
        std::string* physical_property =
            PropertyPolicyHelper::find_property(mp_RTPSParticipant->getAttributes().properties, physical_property_name);
        if (nullptr != physical_property) {
            (void)participant_data->m_properties.push_back(physical_property_name, *physical_property);
        }
    }
}

static void set_builtin_matched_allocation(ResourceLimitedContainerConfig& allocation,
                                           const RTPSParticipantAttributes& pattr) {
    // Matched endpoints will depend on total number of participants
    allocation = pattr.allocation.participants;

    // As participants allocation policy includes the local participant, one has to be substracted
    if (allocation.initial > 1) {
        allocation.initial--;
    }
    if ((allocation.maximum > 1) && (allocation.maximum < std::numeric_limits<size_t>::max())) {
        allocation.maximum--;
    }
}

static void set_builtin_endpoint_locators(EndpointAttributes& endpoint, const PDP* pdp,
                                          const BuiltinProtocols* builtin) {
    const RTPSParticipantAttributes& pattr = pdp->getRTPSParticipant()->getAttributes();

    auto part_data = pdp->getLocalParticipantProxyData();
    if (nullptr == part_data) {
        // Local participant data has not yet been created.
        // This means we are creating the PDP endpoints, so we copy the locators from mp_builtin
        endpoint.multicastLocatorList = builtin->m_metatrafficMulticastLocatorList;
        endpoint.unicastLocatorList = builtin->m_metatrafficUnicastLocatorList;
    } else {  //LCOV_EXCL_START
        // Locators are copied from the local participant metatraffic locators
        endpoint.unicastLocatorList.clear();
        for (const Locator_t& loc : part_data->metatraffic_locators.unicast) {
            endpoint.unicastLocatorList.push_back(loc);
        }
        endpoint.multicastLocatorList.clear();
        for (const Locator_t& loc : part_data->metatraffic_locators.multicast) {
            endpoint.multicastLocatorList.push_back(loc);
        }  //LCOV_EXCL_STOP
    }

    // External locators are always taken from the same place
    endpoint.external_unicast_locators = pdp->builtin_attributes().metatraffic_external_unicast_locators;
    endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
}

ReaderAttributes PDP::create_builtin_reader_attributes() const {
    ReaderAttributes attributes;

    const RTPSParticipantAttributes& pattr = getRTPSParticipant()->getAttributes();
    set_builtin_matched_allocation(attributes.matched_writers_allocation, pattr);
    set_builtin_endpoint_locators(attributes.endpoint, this, mp_builtin);

    // Builtin endpoints are always reliable, transient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;

    // Built-in readers never expect inline qos
    // attributes.expects_inline_qos = false;

    return attributes;
}

WriterAttributes PDP::create_builtin_writer_attributes() const {
    WriterAttributes attributes;

    const RTPSParticipantAttributes& pattr = getRTPSParticipant()->getAttributes();
    set_builtin_matched_allocation(attributes.matched_readers_allocation, pattr);
    set_builtin_endpoint_locators(attributes.endpoint, this, mp_builtin);

    // Builtin endpoints are always reliable, transient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;

    return attributes;
}

#if HAVE_SECURITY
void PDP::add_builtin_security_attributes(ReaderAttributes& ratt, WriterAttributes& watt) const {
    const security::ParticipantSecurityAttributes& part_attr = mp_RTPSParticipant->security_attributes();

    ratt.endpoint.security_attributes().is_submessage_protected = part_attr.is_discovery_protected;
    ratt.endpoint.security_attributes().plugin_endpoint_attributes = PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID;

    watt.endpoint.security_attributes().is_submessage_protected = part_attr.is_discovery_protected;
    watt.endpoint.security_attributes().plugin_endpoint_attributes = PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID;

    if (part_attr.is_discovery_protected) {
        security::PluginParticipantSecurityAttributes plugin_part_attr(part_attr.plugin_participant_attributes);

        if (plugin_part_attr.is_discovery_encrypted) {
            ratt.endpoint.security_attributes().plugin_endpoint_attributes |=
                PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
            watt.endpoint.security_attributes().plugin_endpoint_attributes |=
                PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
        }
        if (plugin_part_attr.is_discovery_origin_authenticated) {
            ratt.endpoint.security_attributes().plugin_endpoint_attributes |=
                PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ORIGIN_AUTHENTICATED;
            watt.endpoint.security_attributes().plugin_endpoint_attributes |=
                PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ORIGIN_AUTHENTICATED;
        }
    }
}

#endif  // HAVE_SECURITY

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
